package org.apache.spark.examples

import org.apache.spark.{SparkConf, SparkContext}

object RDDFilterOperations {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("RDD Filter Operations").setMaster("local[1]")
    val sc = new SparkContext(conf)

    // 创建一个包含1-20的RDD
    val numbersRDD = sc.parallelize(1 to 20, 2)
    println("\n=== 原始数据 ===")
    numbersRDD.collect().foreach(x => print(s"$x "))
    println()

    // 1. 基础filter：过滤偶数
    val evenRDD = numbersRDD.filter(_ % 2 == 0)
    println("\n=== 偶数过滤 ===")
    evenRDD.collect().foreach(x => print(s"$x "))
    println()

    // 2. 范围filter：过滤5到15之间的数
    val rangeRDD = numbersRDD.filter(x => x >= 5 && x <= 15)
    println("\n=== 范围过滤(5-15) ===")
    rangeRDD.collect().foreach(x => print(s"$x "))
    println()

    // 3. 多条件filter：过滤能被3整除且大于10的数
    val multiConditionRDD = numbersRDD.filter(x => x % 3 == 0 && x > 10)
    println("\n=== 多条件过滤(被3整除且>10) ===")
    multiConditionRDD.collect().foreach(x => print(s"$x "))
    println()

    // 4. 自定义函数filter
    def isPrime(n: Int): Boolean = {
      if (n <= 1) return false
      if (n == 2) return true
      !(2 to math.sqrt(n).toInt).exists(x => n % x == 0)
    }
    val primeRDD = numbersRDD.filter(isPrime)
    println("\n=== 质数过滤 ===")
    primeRDD.collect().foreach(x => print(s"$x "))
    println()

    // 5. 使用集合filter：过滤出在指定集合中的数
    val targetSet = Set(2, 4, 6, 8, 10)
    val setFilterRDD = numbersRDD.filter(targetSet.contains)
    println("\n=== 集合包含过滤 ===")
    setFilterRDD.collect().foreach(x => print(s"$x "))
    println()

    // 6. 字符串转换后filter：将数字转为字符串后过滤包含"1"的数
    val containsOneRDD = numbersRDD.filter(_.toString.contains("1"))
    println("\n=== 字符串包含'1'过滤 ===")
    containsOneRDD.collect().foreach(x => print(s"$x "))
    println()

    // 7. 复合filter：先过滤偶数，再过滤大于10的数
    val compositeRDD = numbersRDD
      .filter(_ % 2 == 0)
      .filter(_ > 10)
    println("\n=== 复合过滤(偶数且>10) ===")
    compositeRDD.collect().foreach(x => print(s"$x "))
    println()

    // 8. 异常处理filter
    val safeFilterRDD = numbersRDD.filter(x => {
      try {
        if (x > 15) throw new IllegalArgumentException("Too big!")
        true
      } catch {
        case _: IllegalArgumentException => false
      }
    })
    println("\n=== 异常处理过滤(<=15) ===")
    safeFilterRDD.collect().foreach(x => print(s"$x "))
    println()

    // 9. 数学计算filter：过滤平方根大于3的数
    val sqrtFilterRDD = numbersRDD.filter(x => math.sqrt(x) > 3)
    println("\n=== 平方根>3过滤 ===")
    sqrtFilterRDD.collect().foreach(x => print(s"$x "))
    println()

    // 10. 使用Option的filter
    case class NumberInfo(value: Int, description: Option[String])
    val numbersWithInfo = numbersRDD.map(x => 
      NumberInfo(x, if (x % 2 == 0) Some("even") else None)
    )
    val hasDescriptionRDD = numbersWithInfo.filter(_.description.isDefined)
    println("\n=== Option过滤(有描述的数) ===")
    hasDescriptionRDD.collect().foreach(x => 
      println(s"值: ${x.value}, 描述: ${x.description.get}")
    )

    // 暂停以便查看Spark UI
    Thread.sleep(300000)

    sc.stop()
  }
} 